package com.zf.flink.simple;

import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.springframework.stereotype.Component;

@Component
public class MysqlCdcSource {
    public DataStream<String> createSource(StreamExecutionEnvironment env) {
        MySqlSource<String> mySQLSource = MySqlSource.<String>builder()
                .hostname("localhost")
                .port(3306)
                .databaseList("test_db") // 监控的数据库
                .tableList("test_db.t_user") // 监控的表
                .username("root")
                .password("password")
                .deserializer(new JsonDebeziumDeserializationSchema()) // JSON格式输出
                .startupOptions(StartupOptions.initial()) // 初始快照+增量binlog
                .serverId("5400-5403") // 并行度为4时的server-id范围
                .build();

        return env.fromSource(mySQLSource, WatermarkStrategy.noWatermarks(), "MySQL CDC Source");
    }
}